package industry_2024.industry_04.bjyc

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder}
import org.apache.spark.sql.SparkSession

object bjyc01 {
  def main(args: Array[String]): Unit = {
    /*
          1、根据子任务一的结果，建立随机森林（随机森林相关参数可自定义，不做限制），使用子任务一的结果训练随机森林模型，然后
          再将hudi中dwd.fact_machine_learning_data_test（该表字段含义与dwd.fact_machine_learning_data表相
          同，machine_record_state列值为空，表结构自行查看）转成向量，预测其是否报警将结果输出到MySQL数据
          库shtd_industry中的ml_result表中（表结构如下）。在Linux的MySQL命令行中查询出machine_record_id
          为1、8、20、28和36的5条数据，将SQL语句复制并粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号
          下，将执行结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
    ml_result表结构：
    字段	类型	中文含义	备注
    machine_record_id	int	主键
    machine_record_state	double	设备状态	报警为1，其他状态则为0
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第四套卷子报警预测")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  拿到第一题的结果
    val dataFrame=spark.table("xiaoqiu.fact_machine_learning_data")

    /*
        .slice(3, dataFrame.columns.length - 5)：使用 slice 方法来选择列名的子集。
        3：表示从第四列开始（索引从0开始，所以3对应的是第4列）。
        dataFrame.columns.length - 5：表示到倒数第5列结束。
     */
    //  所以下面就是选择表格的第四列持续到倒数第五列的列名，用数组存储
    val cols:Array[String]=dataFrame.columns.slice(3,dataFrame.columns.length-5)
    println("选择进行处理的字段")
    println(cols.mkString(","))



    /*
        1.dataFrame.na.fill(0.0,cols):表示在特定的列(即cols里面的列)里面用0.0填充空值
        2.randomSplit(Array(0.8,0.2))按照比例将dataframe随机分成两份，并且赋值给变量ranting,test,百分之80的
        的训练集和百分之20的测试集
     */
    val Array(ranting,test)=dataFrame.na.fill(0.0,cols)
      .withColumnRenamed("machine_record_state","label")
      .randomSplit(Array(0.8,0.2))


    //  1.模型输入要求：
    //大多数机器学习模型，包括 RandomForestClassifier，要求输入的特征为一个统一的向量格式。特征向量可以将多个特征合并为一个单一的结构，使得模型可以方便地处理和学习。
    //
    val vectorassembler=new VectorAssembler()
      .setInputCols(cols)
      .setOutputCol("features")

    /*
        RandomForestClassifier 是 Spark MLlib 中用于分类任务的一个机器学习模型。具体来说，这个包的主要功能和用途包括：

    随机森林算法：

    随机森林是一种集成学习方法，它通过构建多个决策树并结合它们的预测结果来提高分类性能。每棵树都是在训练数据的不同子集上训练的，从而增强模型的泛化能力。
    分类任务：

    RandomForestClassifier 专门用于处理分类问题，即预测离散的类别标签。例如，可以用来预测客户是否会购买产品、疾病诊断、图像分类等。
    处理特征的能力：

    随机森林可以处理大量的特征，并且对特征的缩放和归一化不敏感，这使得它在实际应用中非常灵活。
    防止过拟合：

    由于随机森林使用了多棵树的投票机制，它相较于单棵决策树更能抵抗过拟合现象，尤其是在数据量较小或特征维度较高的情况下。
    重要性评估：

    随机森林还可以提供特征的重要性评估，帮助了解哪些特征对模型的预测影响最大，从而进行特征选择或分析。
    超参数调优：

    RandomForestClassifier 允许用户通过设置多个超参数（如树的数量、树的深度等）来调整模型性能，这在机器学习模型训练中是一个重要的步骤。
    在你的代码中，创建 RandomForestClassifier 的实例是为了在机器学习流程中使用该模型进行分类任务。这一过程通常包括特征提取、模型训练、模型评估等步骤。
     */
    //  forest:森林       Classifier:分类器
    val classifier:RandomForestClassifier=new RandomForestClassifier()

    //  定义管道，按顺序处理数据
    val pipeline:Pipeline=new Pipeline()
      .setStages(Array(vectorassembler,classifier))

//  通过构建一个超参数网格和使用交叉验证的方式，系统地评估不同的随机森林模型参数组合，以找到最佳配置，从而提高模型的预测性能。
    /*
        创建超参数网格
    val paramMaps:Array[ParamMap]=new ParamGridBuilder()
      .addGrid(classifier.numTrees,Array(10,20,30))
      .addGrid(classifier.maxDepth,Array(5,10,15))
      .addGrid(classifier.impurity,Array("entropy","gini"))
      .build()
    ParamGridBuilder: 用于构建超参数的网格。
    addGrid: 向参数网格中添加不同的参数组合。每个addGrid方法指定一个要调整的参数和一组可能的值。
    classifier.numTrees: 随机森林中的树的数量，这里尝试10、20和30棵树。
    classifier.maxDepth: 每棵树的最大深度，尝试深度为5、10和15。
    classifier.impurity: 用于衡量不纯度的标准，这里使用"entropy"（信息增益）和"gini"（基尼不纯度）。
    build(): 生成参数组合的数组（Array[ParamMap]），每个组合将用于训练模型。
     */
    //   param:参数         grid:网格     builder:构造器
    val paramMaps:Array[ParamMap]=new ParamGridBuilder()
      .addGrid(classifier.numTrees,Array(10,20,30))
      .addGrid(classifier.maxDepth,Array(5,10,15))
      .addGrid(classifier.impurity,Array("entropy","gini"))         //  impurity:杂质
      .build()


    /*
        CrossValidator: Spark中的交叉验证类，用于评估模型的性能并找到最佳的超参数组合。
    setEstimatorParamMaps(paramMaps): 设置要测试的参数组合，即在第一步构建的paramMaps。
    setEvaluator(new MulticlassClassificationEvaluator()): 设置评估器，这里使用多类分类评估器，评估模型的分类性能（如准确率、召回率等）。
    setEstimator(pipeline): 设置要训练的模型管道，这个管道包括特征提取和分类器。
    setParallelism(3): 指定并行度，表示在交叉验证过程中，可以同时运行3个模型训练任务，以提高效率。
     */
    //  cross:交叉          Validator:验证器
    val crossValidator:CrossValidator=new CrossValidator()
      .setEstimatorParamMaps(paramMaps)
      .setEvaluator(new MulticlassClassificationEvaluator())
      .setEstimator(pipeline)
      .setParallelism(3)

    //  训练模型并且保存
    val model:CrossValidatorModel=crossValidator.fit(ranting)

    //  读取需要预测的数据
    val testData=spark.table("dwd.fact_machine_learning_data_test")

    //  转换测试数据为特征向量
    val testVector=vectorassembler.transform(testData)

    //  使用训练好的模型进行预测
    val predictions=model.transform(testData)

    //  选择需要的列
    val result=predictions.select("machine_record_id","prediction")
      .withColumnRenamed("prediction","machine_record_state")


    //  存入mysql
//    result.write.mode("overwrite")
//      .jdbc()




    spark.close()
  }

}
